package com.example.toy.core.table;

import com.example.toy.ToySystem;
import com.example.toy.core.commit.CommitMsg;
import com.example.toy.core.commit.ToyCommitAbstractFactory;
import com.example.toy.core.commit.ToyCommitView;
import com.example.toy.core.conf.ToyTableConfig;
import com.example.toy.core.fs.ToyTableFileView;
import com.example.toy.core.lock.FileLock;
import com.example.toy.core.locker.ToyLock;
import com.example.toy.core.partition.PartitionPathStyle;
import com.example.toy.core.reader.DefaultToyTableReader;
import com.example.toy.core.reader.ToyTableReader;
import com.example.toy.core.record.key.generator.KeyGenerator;
import com.example.toy.core.utils.PathUtils;
import com.example.toy.core.utils.ToyPathDefine;
import com.example.toy.core.version.ToyVersion;
import com.example.toy.core.writer.DefaultToyTableWriter;
import com.example.toy.core.writer.ToyTableWriter;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * FsOfflineToy
 *
 * <p>这是一个多线程安全的类，但是并不是一个不可变对象， 表可能会在不同的线程中，通过refresh变成另一个版本。
 *
 * @author shenbing
 * @date 2021/12/21
 */
@Slf4j
public class FsOfflineToyTable implements ToyTable {

  private final TableIdentifier tableIdentifier;
  private final Path basePath;
  private final ToySystem toySystem;
  private final boolean isRefreshable;
  private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  private final Lock wLock = readWriteLock.writeLock();
  private final Lock rLock = readWriteLock.readLock();

  private FsOfflineTableMeta tableMeta;

  public FsOfflineToyTable(ToySystem toySystem, Path basePath, ToyVersion toyVersion) {
    this(toySystem, basePath, toyVersion, Objects.equals(toyVersion, ToyVersion.LATEST_VERSION));
  }

  public FsOfflineToyTable(
      ToySystem toySystem, Path basePath, ToyVersion toyVersion, boolean refreshable) {
    this.toySystem = toySystem;
    this.basePath = basePath;
    this.isRefreshable = refreshable;

    Path configFilePath = ToyPathDefine.getTableConfigFilePath(basePath);
    ToyTableConfig config = ToyTableConfig.fromCacheOrPath(configFilePath);
    this.tableIdentifier = new TableIdentifier(config.getTableName());

    if (Objects.equals(toyVersion, ToyVersion.LATEST_VERSION)) {
      ToyCommitAbstractFactory toyCommitAbstractFactory =
          toySystem.getSystemFactoryProducer().buildCommitFactory();
      ToyCommitView commitView =
          toyCommitAbstractFactory.getCommitView(
              new TableIdentifier(config.getTableName()), basePath);
      toyVersion = commitView.getLatestSucceedCommit().getVersion();
    }
    this.tableMeta = new FsOfflineTableMeta(this, basePath, toyVersion);
  }

  @Override
  public ToyTableMeta getTableMeta() {
    try {
      rLock.lock();
      return this.tableMeta;
    } finally {
      rLock.unlock();
    }
  }

  @Override
  public TableIdentifier getTableIdentifier() {
    return this.tableIdentifier;
  }

  @Override
  public ToySystem getToySystem() {
    return this.toySystem;
  }

  @Override
  public void refresh() {
    if (!isRefreshable) {
      throw new IllegalStateException("Unsupported to refresh table: " + tableMeta.getTableName());
    }
    try {
      wLock.lock();
      CommitMsg commitMsg = this.tableMeta.getCommitView().getLatestSucceedCommit();
      ToyVersion version = commitMsg.getVersion();
      ToyVersion currentVersion = this.tableMeta.getVersion();
      if (!version.equals(currentVersion)) {
        this.tableMeta = new FsOfflineTableMeta(this, this.basePath, version);
      }
    } finally {
      wLock.unlock();
    }
  }

  @Override
  public boolean isRefreshable() {
    return isRefreshable;
  }

  @Override
  public ToyTable switchTo(ToyVersion version) {
    try {
      rLock.lock();
      if (version.equals(ToyVersion.LATEST_VERSION)) {
        return new FsOfflineToyTable(this.toySystem, this.basePath, version, true);
      } else {
        return new FsOfflineToyTable(this.toySystem, this.basePath, version);
      }
    } finally {
      rLock.unlock();
    }
  }

  @Override
  public ToyTableWriter createWriter(boolean force) {
    try {
      rLock.lock();
      ToyTable table = switchTo(ToyVersion.LATEST_VERSION);
      return new DefaultToyTableWriter(table.getTableMeta());
    } finally {
      rLock.unlock();
    }
  }

  @Override
  public ToyTableReader createReader() {
    try {
      rLock.lock();
      return new DefaultToyTableReader(this.tableMeta);
    } finally {
      rLock.unlock();
    }
  }

  @Override
  public boolean isArchived() {
    try {
      rLock.lock();
      return tableMeta.getTableConfig().isArchived();
    } finally {
      rLock.unlock();
    }
  }

  @SneakyThrows
  @Override
  public void dropPartition(String partitionKey) {
    try {
      wLock.lock();
      ToyLock partitionLock = getPartitionLock(partitionKey);
      try {
        partitionLock.lockQuickly();
        if (isRefreshable) {
          refresh();
        }
        Map<String, Path> partitionPathMap = tableMeta.getPartitionPathMap();
        Path path = partitionPathMap.get(partitionKey);
        FileSystem fs = path.getFileSystem(new Configuration());
        if (fs.exists(path)) {
          fs.delete(path, true);
        }
      } finally {
        partitionLock.unlock();
      }
    } finally {
      wLock.unlock();
    }
  }

  @Override
  public ToyVersion compact() {
    return null;
  }

  @SneakyThrows
  @Override
  public void drop() {
    try {
      wLock.lock();
      FileSystem fileSystem = tableMeta.getBasePath().getFileSystem(new Configuration());

      ToyLock tableLock = getTableLock();
      try {
        tableLock.lockQuickly();
        if (isRefreshable) {
          refresh();
        }

        Path dataDir = tableMeta.getDataDir();
        if (PathUtils.isSamePath(dataDir, tableMeta.getBasePath())) {
          ToyTableFileView realtimeTableFileView = tableMeta.getRealtimeFileView();
          List<FileStatus> partitionDirs = realtimeTableFileView.getPartitionDirs();
          partitionDirs.forEach(
              p -> {
                try {
                  fileSystem.delete(p.getPath(), true);
                } catch (IOException e) {
                  log.warn("Error to delete partition path: " + p.getPath());
                }
              });
        } else {
          // 直接删除所有的数据文件
          fileSystem.delete(dataDir, true);
        }

        // 删除meta dir
        Path metaDirPath = ToyPathDefine.getMetaDirPath(tableMeta.getBasePath());
        fileSystem.delete(metaDirPath, true);

      } finally {
        tableLock.unlock();
      }
    } finally {
      wLock.unlock();
    }
  }

  @Override
  public ToyLock getTableLock() {
    return new FileLock(ToyPathDefine.getTableLockPath(tableMeta.getBasePath()));
  }

  @SneakyThrows
  @Override
  public ToyLock getPartitionLock(String partitionKey) {
    KeyGenerator partitionKeyGenerator = tableMeta.getTableConfig().getPartitionKeyGenerator();
    List<Object> partitionValues =
        partitionKeyGenerator.fromKey(tableMeta.getPartitionSchema(), partitionKey);

    PartitionPathStyle partitionPathStyle = tableMeta.getPartitionPathStyle();
    Path partitionPath =
        partitionPathStyle.generate(tableMeta.getPartitionSchema(), partitionValues.toArray());

    FileSystem fileSystem = partitionPath.getFileSystem(new Configuration());
    if (!fileSystem.exists(partitionPath)) {
      fileSystem.mkdirs(partitionPath);
    }

    return new FileLock(ToyPathDefine.getPartitionLockPath(partitionPath));
  }

  @Override
  public List<String> getPartitionKeys() {
    return null;
  }

  @Override
  public void close() throws IOException {}

  @Override
  public String toString() {
    return "FsOfflineToyTable { name = " + getTableMeta().getTableName() + " }";
  }
}
